/**
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance
 * with the License.  You may obtain a copy of the License at
 *
 * http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package org.apache.hadoop.hbase.client;

import java.util.Arrays;

import org.apache.commons.lang3.ArrayUtils;
import org.apache.hadoop.hbase.HConstants;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.yetus.audience.InterfaceAudience;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

@InterfaceAudience.Private
public class RegionInfoBuilder {
    private static final Logger LOG = LoggerFactory.getLogger(RegionInfoBuilder.class);

    /** A non-capture group so that this can be embedded. */
    public static final String ENCODED_REGION_NAME_REGEX = "(?:[a-f0-9]+)";

    private static final int MAX_REPLICA_ID = 0xFFFF;

    //TODO: Move NO_HASH to HStoreFile which is really the only place it is used.
    public static final String NO_HASH = null;

    /**
     * RegionInfo for first meta region
     * You cannot use this builder to make an instance of the {@link #FIRST_META_REGIONINFO}.
     * Just refer to this instance. Also, while the instance is actually a MutableRI, its type is
     * just RI so the mutable methods are not available (unless you go casting); it appears
     * as immutable (I tried adding Immutable type but it just makes a mess).
     */
    // TODO: How come Meta regions still do not have encoded region names? Fix.
    // hbase:meta,,1.1588230740 should be the hbase:meta first region name.
    public static final RegionInfo FIRST_META_REGIONINFO = new MutableRegionInfo(1L, TableName.META_TABLE_NAME, RegionInfo.DEFAULT_REPLICA_ID);

    private final TableName tableName;
    private byte[] startKey = HConstants.EMPTY_START_ROW;
    private byte[] endKey = HConstants.EMPTY_END_ROW;
    private long regionId = System.currentTimeMillis();
    private int replicaId = RegionInfo.DEFAULT_REPLICA_ID;
    private boolean offLine = false;
    private boolean split = false;
    private byte[] regionName = null;
    private String encodedName = null;

    public static RegionInfoBuilder newBuilder(TableName tableName) {
        return new RegionInfoBuilder(tableName);
    }

    public static RegionInfoBuilder newBuilder(RegionInfo regionInfo) {
        return new RegionInfoBuilder(regionInfo);
    }

    private RegionInfoBuilder(TableName tableName) {
        this.tableName = tableName;
    }

    private RegionInfoBuilder(RegionInfo regionInfo) {
        this.tableName = regionInfo.getTable();
        this.startKey = regionInfo.getStartKey();
        this.endKey = regionInfo.getEndKey();
        this.offLine = regionInfo.isOffline();
        this.split = regionInfo.isSplit();
        this.regionId = regionInfo.getRegionId();
        this.replicaId = regionInfo.getReplicaId();
        this.regionName = regionInfo.getRegionName();
        this.encodedName = regionInfo.getEncodedName();
    }

    public RegionInfoBuilder setStartKey(byte[] startKey) {
        this.startKey = startKey;
        return this;
    }

    public RegionInfoBuilder setEndKey(byte[] endKey) {
        this.endKey = endKey;
        return this;
    }

    public RegionInfoBuilder setRegionId(long regionId) {
        this.regionId = regionId;
        return this;
    }

    public RegionInfoBuilder setReplicaId(int replicaId) {
        this.replicaId = replicaId;
        return this;
    }

    public RegionInfoBuilder setSplit(boolean split) {
        this.split = split;
        return this;
    }

    public RegionInfoBuilder setOffline(boolean offLine) {
        this.offLine = offLine;
        return this;
    }

    public RegionInfo build() {
        return new MutableRegionInfo(tableName, startKey, endKey, split, regionId, replicaId, offLine, regionName, encodedName);
    }

    /**
     * An implementation of RegionInfo that adds mutable methods so can build a RegionInfo instance.
     */
    @InterfaceAudience.Private
    static class MutableRegionInfo implements RegionInfo, Comparable<RegionInfo> {
        /**
         * The new format for a region name contains its encodedName at the end.
         * The encoded name also serves as the directory name for the region
         * in the filesystem.
         *
         * New region name format:
         *    &lt;tablename>,,&lt;startkey>,&lt;regionIdTimestamp>.&lt;encodedName>.
         * where,
         *    &lt;encodedName> is a hex version of the MD5 hash of
         *    &lt;tablename>,&lt;startkey>,&lt;regionIdTimestamp>
         *
         * The old region name format:
         *    &lt;tablename>,&lt;startkey>,&lt;regionIdTimestamp>
         * For region names in the old format, the encoded name is a 32-bit
         * JenkinsHash integer value (in its decimal notation, string form).
         *<p>
         * **NOTE**
         *
         * The first hbase:meta region, and regions created by an older
         * version of HBase (0.20 or prior) will continue to use the
         * old region name format.
         */

        // This flag is in the parent of a split while the parent is still referenced by daughter
        // regions. We USED to set this flag when we disabled a table but now table state is kept up in
        // zookeeper as of 0.90.0 HBase. And now in DisableTableProcedure, finally we will create bunch
        // of UnassignProcedures and at the last of the procedure we will set the region state to
        // CLOSED, and will not change the offLine flag.
        private boolean offLine = false;
        private boolean split = false;
        private final long regionId;
        private final int replicaId;
        private final byte[] regionName;
        private final byte[] startKey;
        private final byte[] endKey;
        private final int hashCode;
        private final String encodedName;
        private final byte[] encodedNameAsBytes;
        private final TableName tableName;

        private static int generateHashCode(final TableName tableName, final byte[] startKey, final byte[] endKey, final long regionId,
                final int replicaId, boolean offLine, byte[] regionName) {
            int result = Arrays.hashCode(regionName);
            result = (int) (result ^ regionId);
            result ^= Arrays.hashCode(checkStartKey(startKey));
            result ^= Arrays.hashCode(checkEndKey(endKey));
            result ^= Boolean.valueOf(offLine).hashCode();
            result ^= Arrays.hashCode(tableName.getName());
            result ^= replicaId;
            return result;
        }

        private static byte[] checkStartKey(byte[] startKey) {
            return startKey == null ? HConstants.EMPTY_START_ROW : startKey;
        }

        private static byte[] checkEndKey(byte[] endKey) {
            return endKey == null ? HConstants.EMPTY_END_ROW : endKey;
        }

        private static TableName checkTableName(TableName tableName) {
            if(tableName == null) {
                throw new IllegalArgumentException("TableName cannot be null");
            }
            return tableName;
        }

        private static int checkReplicaId(int regionId) {
            if(regionId > MAX_REPLICA_ID) {
                throw new IllegalArgumentException("ReplicaId cannot be greater than" + MAX_REPLICA_ID);
            }
            return regionId;
        }

        /**
         * Private constructor used constructing MutableRegionInfo for the
         * first meta regions
         */
        private MutableRegionInfo(long regionId, TableName tableName, int replicaId) {
            this(tableName, HConstants.EMPTY_START_ROW, HConstants.EMPTY_END_ROW, false, regionId, replicaId, false,
                    RegionInfo.createRegionName(tableName, null, regionId, replicaId, false));
        }

        MutableRegionInfo(final TableName tableName, final byte[] startKey, final byte[] endKey, final boolean split, final long regionId,
                final int replicaId, boolean offLine, byte[] regionName) {
            this(checkTableName(tableName), checkStartKey(startKey), checkEndKey(endKey), split, regionId, checkReplicaId(replicaId), offLine,
                    regionName, RegionInfo.encodeRegionName(regionName));
        }

        MutableRegionInfo(final TableName tableName, final byte[] startKey, final byte[] endKey, final boolean split, final long regionId,
                final int replicaId, boolean offLine, byte[] regionName, String encodedName) {
            this.tableName = checkTableName(tableName);
            this.startKey = checkStartKey(startKey);
            this.endKey = checkEndKey(endKey);
            this.split = split;
            this.regionId = regionId;
            this.replicaId = checkReplicaId(replicaId);
            this.offLine = offLine;
            if(ArrayUtils.isEmpty(regionName)) {
                this.regionName = RegionInfo.createRegionName(this.tableName, this.startKey, this.regionId, this.replicaId,
                        !this.tableName.equals(TableName.META_TABLE_NAME));
                this.encodedName = RegionInfo.encodeRegionName(this.regionName);
            } else {
                this.regionName = regionName;
                this.encodedName = encodedName;
            }
            this.hashCode = generateHashCode(this.tableName, this.startKey, this.endKey, this.regionId, this.replicaId, this.offLine,
                    this.regionName);
            this.encodedNameAsBytes = Bytes.toBytes(this.encodedName);
        }

        /**
         * @return Return a short, printable name for this region
         * (usually encoded name) for us logging.
         */
        @Override
        public String getShortNameToLog() {
            return RegionInfo.prettyPrint(this.getEncodedName());
        }

        /** @return the regionId */
        @Override
        public long getRegionId() {
            return regionId;
        }


        /**
         * @return the regionName as an array of bytes.
         * @see #getRegionNameAsString()
         */
        @Override
        public byte[] getRegionName() {
            return regionName;
        }

        /**
         * @return Region name as a String for use in logging, etc.
         */
        @Override
        public String getRegionNameAsString() {
            return RegionInfo.getRegionNameAsString(this, this.regionName);
        }

        /** @return the encoded region name */
        @Override
        public String getEncodedName() {
            return this.encodedName;
        }

        @Override
        public byte[] getEncodedNameAsBytes() {
            return this.encodedNameAsBytes;
        }

        /** @return the startKey */
        @Override
        public byte[] getStartKey() {
            return startKey;
        }


        /** @return the endKey */
        @Override
        public byte[] getEndKey() {
            return endKey;
        }

        /**
         * Get current table name of the region
         * @return TableName
         */
        @Override
        public TableName getTable() {
            return this.tableName;
        }

        /**
         * Returns true if the given inclusive range of rows is fully contained
         * by this region. For example, if the region is foo,a,g and this is
         * passed ["b","c"] or ["a","c"] it will return true, but if this is passed
         * ["b","z"] it will return false.
         * @throws IllegalArgumentException if the range passed is invalid (ie. end &lt; start)
         */
        @Override
        public boolean containsRange(byte[] rangeStartKey, byte[] rangeEndKey) {
            if(Bytes.compareTo(rangeStartKey, rangeEndKey) > 0) {
                throw new IllegalArgumentException(
                        "Invalid range: " + Bytes.toStringBinary(rangeStartKey) + " > " + Bytes.toStringBinary(rangeEndKey));
            }

            boolean firstKeyInRange = Bytes.compareTo(rangeStartKey, startKey) >= 0;
            boolean lastKeyInRange = Bytes.compareTo(rangeEndKey, endKey) < 0 || Bytes.equals(endKey, HConstants.EMPTY_BYTE_ARRAY);
            return firstKeyInRange && lastKeyInRange;
        }

        /**
         * Return true if the given row falls in this region.
         */
        @Override
        public boolean containsRow(byte[] row) {
            return Bytes.compareTo(row, startKey) >= 0 && (Bytes.compareTo(row, endKey) < 0 || Bytes.equals(endKey, HConstants.EMPTY_BYTE_ARRAY));
        }

        /** @return true if this region is a meta region */
        @Override
        public boolean isMetaRegion() {
            return tableName.equals(FIRST_META_REGIONINFO.getTable());
        }

        /**
         * @return True if has been split and has daughters.
         */
        @Override
        public boolean isSplit() {
            return this.split;
        }

        /**
         * @param split set split status
         * @return MutableRegionInfo
         */
        public MutableRegionInfo setSplit(boolean split) {
            this.split = split;
            return this;
        }

        /**
         * @return True if this region is offline.
         */
        @Override
        public boolean isOffline() {
            return this.offLine;
        }

        /**
         * The parent of a region split is offline while split daughters hold
         * references to the parent. Offlined regions are closed.
         * @param offLine Set online/offline status.
         * @return MutableRegionInfo
         */
        public MutableRegionInfo setOffline(boolean offLine) {
            this.offLine = offLine;
            return this;
        }

        /**
         * @return True if this is a split parent region.
         */
        @Override
        public boolean isSplitParent() {
            if(!isSplit())
                return false;
            if(!isOffline()) {
                LOG.warn("Region is split but NOT offline: " + getRegionNameAsString());
            }
            return true;
        }

        /**
         * Returns the region replica id
         * @return returns region replica id
         */
        @Override
        public int getReplicaId() {
            return replicaId;
        }

        /**
         * @see java.lang.Object#toString()
         */
        @Override
        public String toString() {
            return "{ENCODED => " + getEncodedName() + ", " + HConstants.NAME + " => '" + Bytes
                    .toStringBinary(this.regionName) + "', STARTKEY => '" + Bytes.toStringBinary(this.startKey) + "', ENDKEY => '" + Bytes
                    .toStringBinary(
                            this.endKey) + "'" + (isOffline() ? ", OFFLINE => true" : "") + (isSplit() ? ", SPLIT => true" : "") + ((replicaId > 0) ? ", REPLICA_ID => " + replicaId : "") + "}";
        }

        /**
         * @param o
         * @see java.lang.Object#equals(java.lang.Object)
         */
        @Override
        public boolean equals(Object o) {
            if(this == o) {
                return true;
            }
            if(o == null) {
                return false;
            }
            if(!(o instanceof RegionInfo)) {
                return false;
            }
            return this.compareTo((RegionInfo) o) == 0;
        }

        /**
         * @see java.lang.Object#hashCode()
         */
        @Override
        public int hashCode() {
            return this.hashCode;
        }

        @Override
        public int compareTo(RegionInfo other) {
            return RegionInfo.COMPARATOR.compare(this, other);
        }

    }
}
